#pragma once

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <thread>
#include <memory>
#include <functional>
#include "logger.h"

class MQClient
{
public:
    using ptr = std::shared_ptr<MQClient>;

    // 消息就绪回调函数
    using MessageCallback = std::function<void(const char *, size_t)>;

    MQClient(const std::string &username, const std::string pwd, const std::string host)
    {
        _loop = EV_DEFAULT;
        // 关联句柄和事件循环
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        // 创建连接和信道
        // amqp://用户名:密码@IP:PORT/
        std::string url = "amqp://" + username + ":" + pwd + "@" + host + "/";
        AMQP::Address addr(url);
        _conn = std::make_unique<AMQP::TcpConnection>(_handler.get(), addr);
        _channel = std::make_unique<AMQP::TcpChannel>(_conn.get());

        // 启动线程进行事件循环
        _thread = std::thread([this]()
                              { ev_run(_loop, 0); });
    }

    // 声明交换机和消息队列
    bool declareComponents(const std::string &exchange_name,
                           const std::string &queue_name,
                           const std::string &routing_key = "routing_key",
                           AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct)
    {
        _channel->declareExchange(exchange_name, echange_type)
            .onSuccess([exchange_name]()
                       { LOG_DEBUG("{} 交换机声明成功", exchange_name); })
            .onError([](const char *msg)
                     { LOG_ERROR("声明交换机失败：{}", msg);
                     return false; });

        _channel->declareQueue(queue_name)
            .onSuccess([exchange_name]()
                       { LOG_DEBUG("{} 队列声明成功", exchange_name); })
            .onError([](const char *msg)
                     { LOG_ERROR("声明队列失败：{}", msg);
                     return false; });

        // 对交换机和队列进行绑定
        _channel->bindQueue(exchange_name, queue_name, routing_key)
            .onSuccess([exchange_name, queue_name, routing_key]()
                       { LOG_DEBUG("{} - {} - {} 绑定成功！", exchange_name, queue_name, routing_key); })
            .onError([exchange_name, queue_name](const char *msg)
                     { LOG_ERROR("{} - {} 绑定失败：{}", exchange_name, queue_name, msg);
                     return false; });

        return true;
    }

    // 发布消息
    bool publish(const std::string &exchange,
                 const std::string &msg,
                 const std::string &routing_key = "routing_key")
    {
        LOG_DEBUG("向交换机 {}-{} 发布消息", exchange, routing_key);
        bool ret = _channel->publish(exchange, routing_key, msg);
        if (ret == false)
        {
            LOG_ERROR("{} 发布消息失败：", exchange);
            return false;
        }
        return true;
    }

    // 消费消息
    bool consume(const std::string &queue, const MessageCallback &callback)
    {
        LOG_DEBUG("开始订阅 {} 队列消息", queue);
        _channel->consume(queue, "consume-tag")
            .onReceived([this, callback](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)
                        { callback(message.body(), message.bodySize());
                        _channel->ack(deliveryTag); })
            .onError([queue](const char *message)
                     { LOG_ERROR("订阅 {} 队列消息失败: {}", queue, message);
                     return false; });

        return true;
    }

    ~MQClient()
    {
        // 销毁时要让事件循环线程退出
        ev_async_init(&_async_watcher, watcher_callback);
        ev_async_start(_loop, &_async_watcher);
        ev_async_send(_loop, &_async_watcher);
        _thread.join();
        _loop = nullptr;
    }

private:
    // 异步通知事件回调函数
    static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
    {
        ev_break(loop, EVBREAK_ALL);
    }

private:
    struct ev_loop *_loop;                        // 事件循环
    struct ev_async _async_watcher;               // 异步通知事件
    std::unique_ptr<AMQP::LibEvHandler> _handler; // libev句柄
    std::unique_ptr<AMQP::TcpConnection> _conn;   // 连接对象
    std::unique_ptr<AMQP::TcpChannel> _channel;   // 信道对象
    std::thread _thread;                          // 进行事件循环线程
};